package com.shux.kafka.consumer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月20日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class DefaultConsumerExceptionHandler implements
		IConsumerExceptionHandler {
	private static  int maxRetryTimes = 3;
	private static List<Class<? extends Throwable>> retryExceptionList = new ArrayList<>();
	static {
		retryExceptionList.add(IOException.class);
		retryExceptionList.add(KafkaConsumerException.class);
		
	}
	public DefaultConsumerExceptionHandler(){
		this(retryExceptionList,maxRetryTimes);
	}
	public DefaultConsumerExceptionHandler(List<Class<? extends Throwable>> retryExceptionList){
		this(retryExceptionList,maxRetryTimes);
	}
	public DefaultConsumerExceptionHandler(int maxRetryTimes){
		this(retryExceptionList,maxRetryTimes);
	}
	public DefaultConsumerExceptionHandler(List<Class<? extends Throwable>> retryExceptionList,int maxRetryTimes){
		this.retryExceptionList = retryExceptionList;
		this.maxRetryTimes = maxRetryTimes;
	}
	@Override
	public boolean retry(Throwable t, int calcount) {
		return calcount < maxRetryTimes || retry(t);
	}
	//如果抛出的异常是指定的异常的话，一直会循环下去，不断尝试发送，不受次数的限制
	private boolean retry(Throwable t) {
		for(Class<? extends Throwable> retryException : retryExceptionList){
			if(isSameThrowable(retryException,t)){
				 break;
			 }
		}
		return false;
	}
	/**
	 * 不断迭代Throwable，直到最后一个
	 * @param retryException
	 * @param currentThrowable
	 * @return
	 */
	private boolean isSameThrowable(Class<? extends Throwable> retryException,Throwable currentThrowable) {
		Throwable tempThrowable = currentThrowable;
		while( tempThrowable != null ) {
			if(retryException.isAssignableFrom(currentThrowable.getClass())){
				return true;
			}
			tempThrowable = tempThrowable.getCause();
		}
		return false;
	}
	

}
